package org.intrasearch.spider;

import com.google.inject.Inject;
import org.intrasearch.core.ArtifactId;
import org.intrasearch.core.ArtifactStorage;
import org.intrasearch.core.InvertedIndex;
import org.intrasearch.core.mapreduce.MapReduce;
import org.intrasearch.core.mapreduce.MapReduceService;

/**
 * PATISDG 10.11.2009
 */
public class SpiderImpl implements Spider
{
    private final ArtifactStorage artifactStorage;
    private final StatusLogger logger;
    private final MapReduce<SpiderMapTaskInput, String, ArtifactId> mapReduce;
    private final InvertedIndex invertedIndex;

    @Inject
    public SpiderImpl(ArtifactStorage artifactStorage, InvertedIndex invertedIndex, StatusLogger logger, MapReduceService mapReduceService, MemoryUsageStatus memoryUsageStatus)
    {
        this.artifactStorage = artifactStorage;
        this.logger = logger;
        this.mapReduce = mapReduceService.mapReduceForKey(SpiderImpl.class);
        this.invertedIndex = invertedIndex;
        mapReduce.setHalfReduceThreshold(32768);
        mapReduce.setReduceTask(new InvertedIndexReduceTask(invertedIndex));
        mapReduce.setMapTask(new SpiderMapTask(logger, artifactStorage));
    }

    public void startIndexing(final VirtualFile dir)
    {
        mapReduce.mapInput(new SpiderMapTaskInput(dir));
        mapReduce.finishAndExecute(new Runnable()
        {
            public void run()
            {
                artifactStorage.flush();
                invertedIndex.flush();
                logger.logStatus("Indexing finished for dir: " + dir);
            }
        });
    }

}
